package com.baishancloud.log.douyin

import com.baishancloud.log.format.scala.Origin
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.configuration.Configuration

import java.util.{Objects, Timer, TimerTask}

/**
 * 延迟数据过滤、以及需求中的过滤，需求1和需求2的过滤条件完全一致
 *
 * @author ziqiang.wang 
 * @date 2022/3/1 14:25
 */
class DataFilter extends RichFilterFunction[Origin] {

  var allowLatest: Long = System.currentTimeMillis() - 24 * 3600 * 1000L

  override def open(parameters: Configuration): Unit = {
    new Timer("更新当前时间", true).schedule(new TimerTask {
      override def run(): Unit = allowLatest = System.currentTimeMillis() - 24 * 3600 * 1000L
    }, 60 * 1000, 60 * 1000)
  }

  override def filter(value: Origin): Boolean = {
    if (value != null) {
      val b: Boolean = value.datetimeOfUTC().toEpochSecond < allowLatest
      val b1: Boolean = value.isEdgeLog
      val b2: Boolean = value.httpCode() >= 200 && value.httpCode() < 300
      val b3: Boolean = value.httpMethod() != null && Objects.equals(value.httpMethod().toLowerCase(), "get")
      b && b1 && b2 && b3
    } else {
      false
    }
  }
}
